﻿using CommonModel.Field;
using CommonModel.Model;
using Confluent.Kafka;
using GrpcPtlSave;
using KafkaUtil;
using KafkaUtil.Consumers;
using KafkaUtil.Integration;
using Maikebing.Data.Taos;
using NewLife.Collections;
using NewLife.Log;
using NewLife.Threading;
using Newtonsoft.Json;
using PtlSave.Model;
using PtlSave.Util;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace PtlSave
{
    /*
     * 消费数据，然后保存
     */
    public class DataDeal
    {
        private KafkaConsumerOptions consumerOptions;   //消费配置信息
        private TaosConnectionStringBuilder builder = null; //涛思数据库连接信息

        private Dictionary<TopicPartition, Task> taskChecks = null; //读取最新偏移量任务集 TopicPartition分区信息，每个分区一个任务
        private Dictionary<TopicPartition, OffsetInfo> offsetInfos = null;  //最新偏移量信息 每个分区单独计算偏移量
        private Dictionary<TopicPartition, DefaultKafkaConsumerProvider> consumerOffset = null; //每个分区一个消费者，分区有返回时才生成

        //用于消费数据，key为maxListenNum最大允许任务数
        private Dictionary<int, Task> tasks = null; //消费数据任务集
        private Dictionary<int, DefaultKafkaConsumerProvider> consumers = null; //消费者集
        //private Dictionary<int, Queue<string>> sqlQueues = null; //结果sql队列 key为paramId
        private Dictionary<int, BlockingCollection<string>> sqlQueues = null; //结果sql队列 key为paramId
        private Dictionary<int, bool> lockFlag = null;

        public int listenNum = 50;   //首次默认启动监听任务数量
        public int maxListenNum = 500;   //最大允许任务数
        public int maxGap = 5000;   //未消费数量阀值，超过则需要新增消费任务、涛思数据库连接、消费者
        private long gapSumPre = 0;   //上一次数据差阈值

        public int saveNums = 0;
        private int minute = 0;   //分钟数计算

        //数据库连接池
        private static TaosConnectionPool _taosConnectionPool;

        //初始线程池数量
        private int poolNum = 5;
        //最大允许线程池数量
        private int maxPoolNum = 1000;
        public int waitingNumPre = 0;
        private int maxWaitNum = 10000;  //允许最大等待保存数据量

        //一条SQL语句总长度不能超过64K 即使64*1024=65536
        private int maxSQLLength = 60000;
        private int maxInsTimes = 5;    //执行数据库操作失败时，重新尝试的最大次数

        /***************GRPC所需数据**********************/
        public Int64 ConsumCount = 0;
        public Int64 StoreCount = 0;
        public Int64 StoreErrCount = 0;

        public Dictionary<string, StoreStatic> storeStatics = new Dictionary<string, StoreStatic>();
        /***************GRPC所需数据**********************/

        private Dictionary<int, TableInfo> tableInfos { get; set; } = null;

        /*
         * consumerOptions kafka配置
         * connString 涛思数据库连接参数
         */
        public DataDeal(KafkaConsumerOptions consumerOptions, TaosConnectionStringBuilder builder)
        {
            this.consumerOptions = consumerOptions;
            this.builder = builder;
            this.tableInfos = Program._dataBaseInit.tableInfos;

            XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 初始化：" + listenNum);
            Init();
        }

        /*
         * 初始化
         */
        private void Init()
        {
            consumers = new Dictionary<int, DefaultKafkaConsumerProvider>();
            //sqlQueues = new Dictionary<int, Queue<string>>();
            sqlQueues = new Dictionary<int, BlockingCollection<string>>();
            lockFlag = new Dictionary<int, bool>();

            tasks = new Dictionary<int, Task>();
            taskChecks = new Dictionary<TopicPartition, Task>();
            consumerOffset = new Dictionary<TopicPartition, DefaultKafkaConsumerProvider>();
            offsetInfos = new Dictionary<TopicPartition, OffsetInfo>();

            poolNum = tableInfos.Count;

            //连接池初始化
            _taosConnectionPool = new TaosConnectionPool
            {
                Name = "TaosConnectionPool",
                ConnectionString = builder.ConnectionString,
                Min = poolNum,  //最小poolNum个连接
                //Max = maxPoolNum,   //最大maxPoolNum个连接
                Max = 0,   //最大个数。默认100，0表示无上限
                IdleTime = 10,  //空闲时间20秒，超过该空闲时间的连接将被销毁释放，直到留下最后2个
                AllIdleTime = 60   //完全空闲时间120秒，把最后2个连接都干掉
            };

            //计算、初始化消费任务
            SetParam("Init");

            //每个paramId分配一个队列
            foreach (int key in tableInfos.Keys)
            {
                //sqlQueues.Add(key, new Queue<string>());
                sqlQueues.Add(key, new BlockingCollection<string>());
                lockFlag.Add(key, false);
            }
            //启动写数据库
            new Thread(DealData).Start();

            XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 初始化 listenNum=" + listenNum + " maxListenNum=" + maxListenNum + " maxGap=" + maxGap);
            XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 初始化完成");

            //计算单位时间插入数据库的数据量
            new TimerX(StaticCalc, "", 60 * 1000, 60 * 1000) { Async = true };
            new TimerX(SetSavePool, "Timer", 60 * 1000, 20 * 1000) { Async = true };

        }

        //定时打印
        private void StaticCalc(Object state)
        {
            if (saveNums > 0 || minute==30)
            {
                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " volume " + saveNums + "/minute");
                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 当前 listenNum=" + listenNum + " poolNum=" + poolNum);
            }
            saveNums = 0;
            minute++;
        }

        /*
         * 计算、初始化消费任务
         */
        private void SetParam(Object state)
        {
            if ("Init".Equals(state.ToString()))
            {
                for (int i = 0; i < listenNum; i++)
                {
                    //这样用，kafka封装类里面默认在OnMessageRecieved后自动提交
                    //consumers.Add(i, new Kafka().GetConsumer(consumerOptions, OnMessageRecieved));
                    //这样用，OnMessageRecieved方法里面就要手工Commit提交
                    consumers.Add(i, new Kafka().GetConsumer(consumerOptions));
                    consumers[i].OnMessageRecieved += OnMessageRecieved;
                    tasks.Add(i, consumers[i].ListenAsync());
                }
            }
            else if ("Timer".Equals(state.ToString()))
            {
                long gapSum = 0;
                //所有分区阀值合并计算
                foreach(var key in offsetInfos.Keys)
                {
                    if (offsetInfos[key].offsetGap > 0)
                    {
                        gapSum += offsetInfos[key].offsetGap;
                    }
                }

                //差大于阀值切任务数小于最大任务数
                if (gapSum > maxGap && listenNum < maxListenNum
                    && ((gapSum >= gapSumPre && gapSumPre > 0) || gapSum > 2 * maxGap))
                    //如果阈值比上次的还大，也就是说差值还在增大（差值如果变小的话，说明当前还处理得过来，无需增加）
                    //或者差值已经大于阈值的三倍
                {
                    //这样用，kafka封装类里面默认在OnMessageRecieved后自动提交
                    //consumers.Add(i, new Kafka().GetConsumer(consumerOptions, OnMessageRecieved));
                    //这样用，OnMessageRecieved方法里面就要手工Commit提交
                    consumers.Add(listenNum, new Kafka().GetConsumer(consumerOptions));
                    consumers[listenNum].OnMessageRecieved += OnMessageRecieved;
                    tasks.Add(listenNum, consumers[listenNum].ListenAsync());
                    XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 当前 listenNum=" + listenNum + " maxListenNum=" + maxListenNum + " maxGap=" + maxGap + " gapSum=" + gapSum);
                    listenNum++;
                }
                gapSumPre = gapSum;
            }
        }

        //计算是否增加线程池数量
        private void SetSavePool(Object state)
        {
            int waitingNum = 0; //等待分配保存任务的数据量
            foreach (int key in sqlQueues.Keys)
            {
                waitingNum += sqlQueues[key].Count;
            }

            if (waitingNum > maxWaitNum && poolNum < maxPoolNum && waitingNum > waitingNumPre && waitingNumPre > 0)
            {
                poolNum++;
                ThreadPool.SetMaxThreads(poolNum, poolNum);

                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 增加处理线程池");
                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 当前 poolNum=" + poolNum + " maxPoolNum=" + maxPoolNum + " waitingNum=" + waitingNum + " maxWaitNum=" + maxWaitNum);
            }

            waitingNumPre = waitingNum;
        }

        /*
         * 回调函数
         */
        private void OnMessageRecieved(RecieveResult recieveResult)
        {
            if (taskChecks.Count == 0)
            {
                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 接收到数据，正在处理");
            }

            //第一次有数据返回时新增并启动任务
            if (!taskChecks.ContainsKey(recieveResult.TopicPartition))
            {
                consumerOffset.Add(recieveResult.TopicPartition, new Kafka().GetConsumer(consumerOptions));
                offsetInfos.Add(recieveResult.TopicPartition, new OffsetInfo());
                taskChecks.Add(recieveResult.TopicPartition,
                    consumerOffset[recieveResult.TopicPartition].QueryWatermarkOffsetsAsync(recieveResult.TopicPartition, 10 * 1000, OnWatermarkOffsetsRecieved));
            }
            //获取topicPartition分区当前偏移量
            offsetInfos[recieveResult.TopicPartition].offset = recieveResult.Offset;

            ConsumCount++;
            if (DataCache(recieveResult))
            {
                recieveResult.Commit();
            }
        }

        /*
         * 定时读取最大偏移量的回调函数
         */
        private void OnWatermarkOffsetsRecieved(TopicPartition topicPartition,WatermarkOffsets watermarkOffsets)
        {
            //获取topicPartition分区最大偏移量
            offsetInfos[topicPartition].offsetMax = (watermarkOffsets.High - 1);

            if (offsetInfos[topicPartition].offsetGap > 0)
            {
                XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " listenNum=" + listenNum + " topicPartition=" + topicPartition.ToInt() +
                    " offsetGap=" + offsetInfos[topicPartition].offsetGap +
                    " offset=" + offsetInfos[topicPartition].offset +
                    " offsetMax=" + offsetInfos[topicPartition].offsetMax);
            }

            //每次返回判断一次
            SetParam("Timer");
        }

        /*
         * 保存数据处理线程
         */
        private void DealData()
        {
            ThreadPool.SetMinThreads(poolNum, poolNum);
            ThreadPool.SetMaxThreads(maxPoolNum, maxPoolNum);
            XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 线程启动");
            while (true)
            {
                try
                {
                    foreach (int key in sqlQueues.Keys)
                    {
                        if (!lockFlag[key] && sqlQueues[key].Count > 0)
                        {
                            lockFlag[key] = true;
                            ThreadPool.QueueUserWorkItem(SavePool, key);
                        }
                        Thread.Sleep(0);
                    }
                }
                catch { }
                finally
                {
                    Thread.Sleep(0);
                }
            }
        }

        /*
         * 保存数据拼接
         */
        private void SavePool(Object data)
        {
            int key = int.Parse(data.ToString());
            //Queue<string> sqlCache = sqlQueues[key];
            BlockingCollection<string> sqlCache = sqlQueues[key];
            string insHead = "INSERT INTO";
            string sql = "";
            string tmpSql = "";
            try
            {
                int cnt = 0;
                while (sqlCache.Count > 0)
                {
                    lock (sqlCache)
                    {
                        tmpSql = sqlCache.Take();
                    }

                    /*
                     * 创建表
                     */
                    if (tmpSql.ToUpper().Contains("CREATE TABLE"))
                    {
                        Execute(tmpSql);
                    }
                    else
                    { 
                        sql = sql + " " + tmpSql;
                        cnt++;
                        if (sql.Length > maxSQLLength)
                        {
                            ThreadPool.QueueUserWorkItem(SavePoolIns, new Object[] { key, cnt, insHead + sql.ToString() });
                            sql = "";
                            cnt = 0;
                        }
                    }
                    Thread.Sleep(1);
                }
                if (!"".Equals(sql))
                {
                    ThreadPool.QueueUserWorkItem(SavePoolIns, new Object[] { key, cnt, insHead + sql.ToString() });
                }
            }
            catch { }
            finally
            {
                lockFlag[key] = false;
            }
        }

        private bool Execute(string sql, int times = 0)
        {
            bool rs = false;
            TaosConnection taosDbConnection = _taosConnectionPool.Get();
            try
            {
                ////执行sql，写入数据库
                if (taosDbConnection.State != System.Data.ConnectionState.Open)
                {
                    taosDbConnection.Open();
                }
                taosDbConnection.CreateCommand(sql).ExecuteNonQuery();
                rs = true;
            }
            catch (Exception e)
            {
                if (times == maxInsTimes)
                {
                    XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 执行[" + sql + "]失败，" + e.Message);
                }
                rs = times < maxInsTimes ? Execute(sql, ++times) : false;
            }
            finally
            {
                _taosConnectionPool.Put(taosDbConnection);
            }
            return rs;
        }

        private void SavePoolIns(Object data)
        {
            SaveToDb(data);
        }

        //保存数据到数据库
        private bool SaveToDb(Object data, int times = 0)
        {
            bool rs = true;
            int key = int.Parse(((Object[])data)[0].ToString());
            int cnt = int.Parse(((Object[])data)[1].ToString());
            string sql = ((Object[])data)[2].ToString();
            TaosConnection taosDbConnection = _taosConnectionPool.Get();
            try
            {
                int affectNum = 0;
                if (taosDbConnection.State != System.Data.ConnectionState.Open)
                {
                    taosDbConnection.Open();
                }
                affectNum = taosDbConnection.CreateCommand(sql).ExecuteNonQuery();

                if (affectNum == 0)
                {
                    rs = false;
                    StoreErrCount = StoreErrCount + cnt;
                    XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 写数据库失败！" + " sql:" + sql);
                }
                else
                {
                    saveNums += affectNum;
                    StoreCount += affectNum;
                    storeStatics[tableInfos[key].tableName].StoreCount = storeStatics[tableInfos[key].tableName].StoreCount + ulong.Parse(affectNum.ToString());
                    //XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " " + affectNum + "条记录写数据库成功！");

                    if (cnt != affectNum)
                    {
                        StoreErrCount = StoreErrCount + (cnt - affectNum);
                    }
                }
            }
            catch (Exception e)
            {
                if (times == maxInsTimes)
                {
                    XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 插入数据失败！" + e.Message + " " + sql);
                }
                rs = times < maxInsTimes ? SaveToDb(data, ++times) : false;
            }
            finally
            {
                _taosConnectionPool.Put(taosDbConnection);
            }

            return rs;
        }

        /*
         * 数据处理
         */
        private bool DataCache(RecieveResult recieveResult)
        {
            bool rs = true;
            StringBuilder sql = new StringBuilder();
            try
            {
                int paramId = -1;
                Dictionary<string, Object> data = null;
                try
                {
                    //存在报错的可能
                    data = JsonConvert.DeserializeObject<Dictionary<string, Object>>(recieveResult.Message.ToString());
                }
                catch
                {
                    data = new Dictionary<string, object>();
                }
                Dictionary<string, Object> addr = null;
                if (data.ContainsKey("addr"))
                {
                    addr = JsonConvert.DeserializeObject<Dictionary<string, Object>>(data["addr"].ToString());
                    paramId = addr.ContainsKey("paramId") ? int.Parse(addr["paramId"].ToString()) : paramId;
                }
                if (paramId != -1 && addr != null)
                {
                    if (tableInfos.ContainsKey(paramId))
                    {
                        if (!storeStatics.ContainsKey(tableInfos[paramId].tableName))
                        {
                            storeStatics.Add(tableInfos[paramId].tableName, new StoreStatic { TableName = tableInfos[paramId].tableName, ConsumCount = 1, StoreCount = 0 });
                        }
                        else
                        {
                            storeStatics[tableInfos[paramId].tableName].ConsumCount = storeStatics[tableInfos[paramId].tableName].ConsumCount + 1;
                        }

                        string ip = addr.ContainsKey("ip") ? addr["ip"].ToString() : "127.0.0.1";
                        string port = addr.ContainsKey("port") ? addr["port"].ToString() : "0";
                        string tableName = tableInfos[paramId].tableName + "_" + ip.Replace(".", "_") + "_" + port;
                        string schema_name = addr.ContainsKey("schema_name") ? addr["schema_name"].ToString() : "";

                        //表不存在，则创建
                        if (!tableInfos[paramId].tableNames.ContainsKey(tableName.ToUpper()))
                        {
                            XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 需要创建表! " + tableName);
                            tableInfos[paramId].tableNames.Add(tableName.ToUpper(), 0);
                            //写入队列
                            lock (sqlQueues[paramId])
                            {
                                //创建表语句一并放入队列
                                //sqlQueues[paramId].Enqueue(tableInfos[paramId].GetCreateTableSql(tableName, ip, port));
                                sqlQueues[paramId].Add(tableInfos[paramId].GetCreateTableSql(tableName, ip, port));
                            }
                        }

                        /*************************语句拼接*************************************************/

                        if (data.ContainsKey("datas"))
                        {
                            //GetInsSql可能会修改tableInfos[paramId]的数据，所以放在前面
                            string vals = GetInsSql(paramId, data["datas"].ToString());

                            sql.Clear();
                            //数据库字段 即配置的字段
                            sql.Append(tableName).Append("(").Append(tableInfos[paramId].columnsStr).Append(")")
                                //时间戳ts
                                .Append(" values('").Append(HiResDateTime.GetCurrentTime(tableInfos[paramId].tableName)).Append("'")
                                //数据来源c_from ip:port
                                .Append(",'").Append(ip).Append(":")
                                .Append(port).Append("','")
                                //c_paramId
                                .Append(paramId.ToString()).Append("','")
                                .Append(schema_name).Append("'");
                            sql.Append(vals);
                            sql.Append(")");
                            //执行sql，写入数据库
                            //lock (taosCommands[n])
                            //{
                            //    taosCommands[n].CommandText = sql.ToString();
                            //    affectNum = taosCommands[n].ExecuteNonQuery();
                            //}

                            //写入队列
                            lock (sqlQueues[paramId])
                            {
                                //sqlQueues[paramId].Enqueue(sql.ToString());
                                sqlQueues[paramId].Add(sql.ToString());
                            }
                            //SaveToDb(new Random().Next(0, poolNum), "insert into " + sql.ToString());
                        }
                        else
                        {
                            XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 参数datas缺失，请检查！");
                        }

                        /*************************语句拼接*************************************************/
                    }
                    else
                    {
                        XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 配置参数缺失，请检查MySQL连接！");
                    }
                }
                else
                {
                    XTrace.WriteLine(CommonUtil.GetFormatDatetime() + " 参数ptlinfo-paramId缺失，请检查！");
                }
            }
            catch (Exception e)
            {
                XTrace.WriteLine("生成保存sql失败！" + e.Message + " sql:" + sql.ToString() + " 消费数据 " + recieveResult.Message);
                rs = false;
            }
            return rs;
        }

        /*
         * sql语句拼接
         */
        private string GetInsSql(int paramId, string dataStr)
        {
            StringBuilder sql = new StringBuilder();
            try
            {
                Dictionary<string, Object> datas = JsonConvert.DeserializeObject<Dictionary<string, Object>>(dataStr);
                Dictionary<int, Object> dataPre = new Dictionary<int, object>();
                for (int i = 0; i <= tableInfos[paramId].layerNum; i++)
                {
                    dataPre.Add(i, null);
                }

                //配置的字段 循环获取值
                foreach (int key in tableInfos[paramId].columns.Keys)
                {
                    int layer = tableInfos[paramId].columns[key].layer;
                    string col = tableInfos[paramId].columns[key].columnName;
                    string type = tableInfos[paramId].columns[key].columnType;
                    string loop = ((dynamic)tableInfos[paramId].columns[key].field).getLoop();

                    //动态字段判断处理
                    string dynamicVal = "";
                    bool updFlag = false;   //是否需要更新
                    if (tableInfos[paramId].dynamic.ContainsKey(col))
                    {
                        dynamicVal = tableInfos[paramId].dynamic[col];
                    }

                    //是末端节点
                    if (tableInfos[paramId].columns[key].field is FieldSchema)
                    {
                        if (layer == 0)
                        {
                            dataPre[layer] = datas.ContainsKey(col) ? datas[col].ToString() : "";
                            sql.Append(GetVal(type, dataPre[layer].ToString()));

                            //dynamicVal非空（说明是动态字段）
                            //val是数字类型
                            //数值比较 dynamicVal小于val
                            if (!"".Equals(dynamicVal) && CommonUtil.IsInt(dataPre[layer].ToString()) && int.Parse(dynamicVal) < int.Parse(dataPre[layer].ToString()))
                            {
                                dynamicVal = dataPre[layer].ToString();
                                updFlag = true;
                            }
                        }
                        else
                        {
                            if (!"".Equals(dataPre[layer - 1].ToString()))
                            {
                                try
                                {
                                    //数组
                                    List<Dictionary<string, Object>> dataIn = JsonConvert.DeserializeObject<List<Dictionary<string, Object>>>(dataPre[layer - 1].ToString());
                                    string val = "";
                                    if (dataIn.Count > tableInfos[paramId].columns[key].loopIdx)
                                    {
                                        val = dataIn[tableInfos[paramId].columns[key].loopIdx].ContainsKey(col) ? dataIn[tableInfos[paramId].columns[key].loopIdx][col].ToString() : val;
                                    }
                                    sql.Append(GetVal(type, val));

                                    //dynamicVal非空（说明是动态字段）
                                    //val是数字类型
                                    //数值比较 dynamicVal小于val
                                    if (!"".Equals(dynamicVal) && CommonUtil.IsInt(val) && int.Parse(dynamicVal) < int.Parse(val))
                                    {
                                        dynamicVal = val;
                                        updFlag = true;
                                    }
                                }
                                catch
                                {
                                    sql.Append(",NULL");
                                }
                            }
                            else
                            {
                                sql.Append(",NULL");
                            }
                        }
                    }
                    //父节点只负责给dataPre赋值
                    else if (tableInfos[paramId].columns[key].field is SchemaDictionary)
                    {
                        if (layer == 0)
                        {
                            dataPre[layer] = datas.ContainsKey(col) ? datas[col].ToString() : "";
                        }
                        else
                        {
                            if (!"".Equals(dataPre[layer - 1].ToString()))
                            {
                                try
                                {
                                    //数组
                                    List<Dictionary<string, Object>> dataIn = JsonConvert.DeserializeObject<List<Dictionary<string, Object>>>(dataPre[layer - 1].ToString());
                                    string val = "";
                                    if (dataIn.Count > tableInfos[paramId].columns[key].loopIdx)
                                    {
                                        val = dataIn[tableInfos[paramId].columns[key].loopIdx].ContainsKey(col) ? dataIn[tableInfos[paramId].columns[key].loopIdx][col].ToString() : val;
                                    }
                                    dataPre[layer] = val;
                                }
                                catch
                                {
                                    dataPre[layer] = "";
                                }
                            }
                            else
                            {
                                dataPre[layer] = "";
                            }
                        }
                    }

                    //需要更新
                    if (updFlag)
                    {
                        //更新表字段逻辑
                        //应该包含写表当前最大val的逻辑，否则下次还要执行一次
                        lock (tableInfos[paramId])
                        {
                            Program._dataBaseInit.UpdCfg(paramId, col, dynamicVal);
                        }
                        //更新变量后，直接调用自身并返回
                        return GetInsSql(paramId, dataStr);
                    }
                }
            }
            catch
            {
                sql.Clear();
                foreach (int key in tableInfos[paramId].columns.Keys)
                {
                    //非父节点才有值
                    if (!tableInfos[paramId].columns[key].isParent)
                    {
                        sql.Append(",NULL");
                    }
                }
            }
            return sql.ToString();
        }

        private string GetVal(string type, string val)
        {
            string rs = "";
            if ("FLOAT".Equals(type) || "INT".Equals(type) || "BIGINT".Equals(type))
            {
                if (CommonUtil.IsInt(val))
                {
                    rs = "," + val;
                }
                else
                {
                    rs = ",NULL";
                }
            }
            else
            {
                rs = "," + ("".Equals(val) ? "NULL" : "'" + val + "'");
            }
            return rs;
        }
    }

    /*
     * 实时分区当前消费偏移量、最大偏移量及偏移量差值信息
     */
    public class OffsetInfo 
    {
        public long offset = 0;
        public long offsetMax = 0;
        public long offsetGap
        {
            get 
            { 
                return offsetMax - offset;
            } 
        }
    }
}
